/*
 * 文件名：KafKaRealMessageSource.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：napo
 * 修改时间：2014-12-25
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.protocal.realmq;

import java.util.Properties;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.realstream.server.monitor.EventCounter;
import com.google.common.base.Preconditions;

/**
 * 
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2014-12-25]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class KafKaRealMessageSource extends AbstractPollableSource
{
	private static final Logger logger = LoggerFactory.getLogger(KafKaRealMessageSource.class);
	private EventCounter eventCounter;
	private String dataType;
	private String zkConnect;
	private String consumerGroupid;
	private int numsPatitions;
	private KafKaMessageConsumer consumer;
	private String topicName;
	private int errorThreshold = 10;

	private int jmsExceptionCounter;

	public KafKaRealMessageSource(EventCounter eventCounter)
	{
		super();
		this.eventCounter = eventCounter;
	}

	/**
	 * @see com.ailk.bigdata.etl.restful.realmq.AbstractPollableSource#doConfigure(com.ailk.bigdata.etl.restful.realmq.properties)
	 */
	@Override
	protected void doConfigure(Properties properties) throws MessgaeDataAcceptException
	{
		this.dataType = properties.getProperty(JMSSourceConfiguration.KAFKA_CONSUMER_INTERFACE_TYPE);
		this.zkConnect = properties.getProperty(JMSSourceConfiguration.KAFKA_ZOOKEEPER_CONNECT);
		this.consumerGroupid = properties.getProperty(JMSSourceConfiguration.KAFKA_CONSUMER_GROUP_ID, "etl_agent_group");
		this.topicName = properties.getProperty(JMSSourceConfiguration.KAFKA_TOPIC_NAME);
		String numsPatition = properties.getProperty(JMSSourceConfiguration.KAFKA_NUMS_PATITION);

		Preconditions.checkArgument(StringUtils.isNotEmpty(dataType), "接口类型不能为空........................");
		Preconditions.checkArgument(StringUtils.isNotEmpty(zkConnect), "zookeeper连接串不能为空........................");

		this.numsPatitions = Integer.parseInt(numsPatition);
	}

	/**
	 * @see com.ailk.bigdata.etl.restful.realmq.AbstractPollableSource#doStart()
	 */
	@Override
	protected void doStart() throws MessgaeDataAcceptException
	{
		try
		{
			consumer = createConsumer();
			jmsExceptionCounter = 0;
		}
		catch (Exception e)
		{
			logger.error("Unable to create consumer", e);
			throw new MessgaeDataAcceptException("Unable to create consumer", e);
		}
	}

	/**
	 * @see com.ailk.bigdata.etl.restful.realmq.AbstractPollableSource#doStop()
	 */
	@Override
	protected void doStop() throws MessgaeDataAcceptException
	{
		if (consumer != null)
		{
			consumer.close();
		}
	}

	/**
	 * @see com.ailk.bigdata.etl.restful.realmq.AbstractPollableSource#doProcess()
	 */
	@Override
	protected Status doProcess() throws Exception
	{
		try
		{
			if (consumer == null)
			{
				consumer = createConsumer();
			}
			return consumer.take();
		}
		catch (Exception jmsException)
		{
			logger.warn("JMSException consuming events", jmsException);
			if (++jmsExceptionCounter > errorThreshold)
			{
				if (consumer != null)
				{
					logger.warn("Exceeded JMSException threshold, closing consumer");
					consumer.close();
					consumer = null;
				}
			}
		}
		catch (Throwable throwable)
		{
			logger.error("Unexpected error processing events", throwable);
			if (throwable instanceof Error)
			{
				throw (Error) throwable;
			}
		}
		finally
		{
			if (consumer != null)
			{
				consumer.commit();
				jmsExceptionCounter = 0;
			}
		}
		return Status.BACKOFF;
	}

	private KafKaMessageConsumer createConsumer() throws Exception
	{
		logger.info("Creating new consumer for " + topicName);
		KafKaMessageConsumer consumer = new KafKaMessageConsumer(eventCounter, zkConnect, consumerGroupid, numsPatitions, topicName, dataType);
		jmsExceptionCounter = 0;
		return consumer;
	}

}
